#pragma once
#include "buffer.hpp"
/*
异步工作器模块:
功能:异步处理主线程的日志消息，完成日志消息的实际落地
提供接口:
stop();//停止异步工作器
push();//方便工作线程向异步工作器提供数据
成员:
1、双缓冲区(生&消
2、互斥锁(保护生产者&生产者 消费者&生产者)
3、双条件变量(双缓冲区都为空,则阻塞异步线程;若生产缓冲区不够了，阻塞生产者)
4、异步线程
5、停止标志
6、实际处理逻辑(具体如何实现交给外部来决定，实现解耦)
*/
#include <atomic>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
namespace LoggerSpace
{
    using func_t = std::function<void(Buffer &)>;
    class AsyncLooper
    {
    public:
        AsyncLooper(const func_t &f, bool safe = true) : _work(f), _stop(false), _t(&AsyncLooper::threadRuning, this), _safe(safe),_count(0)
        {}
        ~AsyncLooper()
        {
            stop();
        }
        void stop()
        {
            _stop = true;
             _conCond.notify_all();
            _t.join();
        }
        void push(const char *str, int len)
        {
            if (_stop)
                return;
            {
                std::unique_lock<std::mutex> locker(_mux);
                // 唤醒条件:或者实际写入的长度小于可写入的长度
                // 安全模式下执行
                if (_safe)
                {
                    //空间不足,将自己进行阻塞，阻塞线程数++
                    //当被唤醒过后,阻塞线程数--
                    //使用while，避免虚假唤醒...
                    while(len> _proBuf.AbleWriteSize())
                    {
                        _count+=1;
                        _proCond.wait(locker);
                        _count-=1;
                    }
                }
                // 非安全模式下直接插入，不考虑空间问题，少了会自动扩容
                _proBuf.write(str, len);
            }
            // 唤醒消费者,有数据来消费了
            _conCond.notify_all();
        }

    private:
        void threadRuning()
        {
            while (true)
            {
                if (_conBuf.empty())
                {
                    std::unique_lock<std::mutex> locker(_mux);
                    _conCond.wait(locker, [&]() -> bool
                                  { return _stop || false == _proBuf.empty(); });
                    if (_stop && _proBuf.empty())
                        break;
                    // 有数据可拿
                    // 交换
                    _conBuf.reset();
                    _proBuf.swap(_conBuf);
                    _proCond.notify_all();
                }
                _work(_conBuf);
            }
            while(_count) _proCond.notify_one();
            _work(_proBuf);
        }
        bool _safe;//控制异步工作器是否是安全模式
        Buffer _proBuf; // 生产缓冲区
        Buffer _conBuf; // 消费缓冲区
        std::mutex _mux;
        std::condition_variable _proCond; // 生产者条件队列
        std::condition_variable _conCond; // 消费者条件队列
        std::atomic<bool> _stop;          // 停止标志
        std::thread _t;                   // 异步线程
        func_t _work;                     // 实际处理逻辑
        std::atomic<int> _count;//记录在生产者中的阻塞队列中的生产者线程数
    };
} 
